package com.bjbn.service;

import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * Kafka消息发送服务
 */
@Service
@Slf4j
public class KafkaService {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    // 定义Kafka主题常量
    public static final String TOPIC_CHARGE_RECORD = "T-CAN-TRAFFIC-CENTER-CHARGE-RECORD";
    public static final String TOPIC_EXIT_INFO = "T-CAN-TRAFFIC-CENTER-EXIT-INFO";
    public static final String TOPIC_ENTRY_INFO = "T-CAN-TRAFFIC-CENTER-ENTRY-INFO";
    public static final String TOPIC_PARKING_SPACE_STATUS = "T-CAN-TRAFFIC-CENTER-PARKING-SPACE-STATUS";
    public static final String TOPIC_INDOOR_PARKING = "T-CAN-TRAFFIC-CENTER-INDOOR-PARKING";
    public static final String TOPIC_OUTDOOR_PARKING = "T-CAN-TRAFFIC-CENTER-OUTDOOR-PARKING";

    /**
     * 发送收费记录到Kafka
     */
    public void sendChargeRecord(Object data) {
        sendMessage(TOPIC_CHARGE_RECORD, data);
    }

    /**
     * 发送出场信息到Kafka
     */
    public void sendExitInfo(Object data) {
        sendMessage(TOPIC_EXIT_INFO, data);
    }

    /**
     * 发送入场信息到Kafka
     */
    public void sendEntryInfo(Object data) {
        sendMessage(TOPIC_ENTRY_INFO, data);
    }

    /**
     * 发送车位状态到Kafka
     */
    public void sendParkingSpaceStatus(Object data) {
        sendMessage(TOPIC_PARKING_SPACE_STATUS, data);
    }

    /**
     * 发送室内停车位信息到Kafka
     */
    public void sendIndoorParking(Object data) {
        sendMessage(TOPIC_INDOOR_PARKING, data);
    }

    /**
     * 发送室外停车位信息到Kafka
     */
    public void sendOutdoorParking(Object data) {
        sendMessage(TOPIC_OUTDOOR_PARKING, data);
    }

    /**
     * 发送消息到指定主题
     */
    private void sendMessage(String topic, Object data) {
        try {
            // 将数据对象转换为JSON字符串
            String jsonMessage = JSON.toJSONString(data);
            log.debug("准备发送消息到Kafka主题 {}，消息内容: {}", topic, jsonMessage);
            kafkaTemplate.send(topic, jsonMessage);

        } catch (Exception e) {
            log.error("发送消息到Kafka异常: {}", e.getMessage(), e);
        }
    }
}